use crate::common::cs;
use crate::base;
use log::{info};
use tokio::task;
use tokio::sync::mpsc;
use crate::common::models::OperationOrder;
use paho_mqtt::Message;
use tokio::sync::mpsc::Sender;
use crate::mqtt::{MQTT_VERSION_5, QOS_2};
use crate::server::control_service;

/// 初始化mqtt
pub async fn init_mqtt() -> cs::Result<()>
{
    println!(" - starting init mqtt");
    connect().await?;
    println!(" - mqtt connected");
    listen_current_topic().await;
    call_server_mqtt().await?;
    println!(" - mqtt init complete");
    Ok(())
}

/// 链接客户端
async fn connect() -> cs::Result<()>
{
    let mut client = base::mqtt_client();
    client.set_mqtt_version(MQTT_VERSION_5);
    client.set_publish_qos(QOS_2);
    client.set_subscribe_qos(QOS_2);
    let resp = client.connect_default().await?;
    info!("mqtt connect : `{:?}`",resp.reason_code());
    Ok(())
}

/// 监听mqtt的当前mac主题
async fn listen_current_topic()
{
    let (s, mut r) = mpsc::channel(1);
    task::spawn(listen_current_topic_loop(s));
    let ss = r.recv().await.unwrap();
    let mut client = base::MQTT_CLIENT.lock().unwrap();
    client.subscribe(base::mqtt_mac(), ss).await.unwrap();
}

async fn listen_current_topic_loop(s: Sender<Sender<Message>>)
{
    let (sender, mut receiver) = mpsc::channel(1);
    s.send(sender).await.unwrap();
    info!("starting listen mqtt topic : `{}`",base::mqtt_mac());
    loop {
        if let Some(msg) = receiver.recv().await {
            if let Ok(s) = String::from_utf8(msg.payload().to_vec()) {
                if let Err(e) = analysis_mqtt_message(&s).await {
                    info!("mqtt message can not analysis:`{}` err:`{}`",s,e);
                }
            }
        };
    }
}

/// 向SERVER TOPIC发送自身的频道
async fn call_server_mqtt() -> cs::Result<()>
{
    if envmnt::is_or("MQTT_INIT_SEND_SERVER", false) {
        let server_name = envmnt::get_or("MQTT_SERVER_TOPIC_NAME", "SERVER");
        let client = base::mqtt_client();
        client.publish(server_name, base::mqtt_mac().as_bytes()).await?;
    }
    Ok(())
}


/// 解析并执行mqtt发送来的消息
async fn analysis_mqtt_message(msg: &str) -> cs::Result<()>
{
    //解析
    let operation_order = serde_json::from_str::<OperationOrder>(msg)?;
    control_service::control_device(operation_order).await?;
    Ok(())
}